package gr.aueb.cs.mscis.chordroid.dht;

import gr.aueb.cs.mscis.chordroid.ChordroidService;
import gr.aueb.cs.mscis.chordroid.app.FileChunk;
import gr.aueb.cs.mscis.chordroid.chord.NodeIdentifier;

import java.io.IOException;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.HashSet;
import java.util.Set;
import java.util.SortedMap;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentSkipListMap;

import android.util.Log;

/**
 * This class is the core of the DHT layer that
 * stands over Chord
 * @author lupin
 *
 */
public class DHTManager implements Runnable {

	private SortedMap<String, FileChunk> dht;
	private NodeIdentifier pred; 
	private NodeIdentifier succ; 
	private NodeIdentifier node;
	
	public DHTManager() {
		// thread-safe implementation
		dht = new ConcurrentSkipListMap<String, FileChunk>();
	}
	
	public String getFirst() {
		if (dht.isEmpty()) return null;
		return dht.firstKey();
	}
	
	public String getLast() {
		if (dht.isEmpty()) return null;
		return dht.lastKey();
	}
	
	public FileChunk getByHash(String hash) {
		if (dht.containsKey(hash)) return dht.get(hash);
		return null;
	}
	
	public Set<String> getKeys() {
		return dht.keySet();
	}
	
	public void addEntry(String hash, FileChunk chunk) {
//		Log.v("DHT", "chunk added: " + hash);
		dht.put(hash, chunk);
	}
	
	private Set<FileChunk> getEntriesForPredecessor() {
		if (ChordroidService.getMyLuckyNode().getPredecessor() == null) return null;

		Set<FileChunk> set = new HashSet<FileChunk>();
		for (Entry<String,FileChunk> e : dht.entrySet()) {
			if (!betweenPredAndMe(e.getKey())) { 
				set.add(e.getValue());
				dht.remove(e.getKey());
			}
		}
		return set;
	}
	
	public void redistributeKeys() {
		pred = ChordroidService.getMyLuckyNode().getPredecessor();
		succ = ChordroidService.getMyLuckyNode().getSuccessor();
		node = ChordroidService.getMyLuckyNode().getNodeId();
		if (dht.isEmpty() || succ.equals(node) || pred == null) return;

		Set<FileChunk> toPredecessorDht = getEntriesForPredecessor();
		if (toPredecessorDht != null) {
			if (!toPredecessorDht.isEmpty()) {
				Log.v("DHT", "sending to pred");
				try {
					Socket toServer = new Socket(pred.getAddr(), pred.getPort() + 2);

					ObjectOutputStream out = new ObjectOutputStream(toServer.getOutputStream());
					out.writeObject(toPredecessorDht);

					toServer.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}			
	}
	
	public void leave() {
		NodeIdentifier succ = ChordroidService.getMyLuckyNode().getSuccessor();
		if (succ.equals(ChordroidService.getMyLuckyNode().getNodeId())) return;
		Set<FileChunk> set = new HashSet<FileChunk>();
		for(Entry<String, FileChunk> e : dht.entrySet()) {
			set.add(e.getValue());
		}
		Log.v("DHT", "leaving. sending chunks: " + set.size());
		try {
			Socket toServer = new Socket(succ.getAddr(), succ.getPort() + 2);

			ObjectOutputStream out = new ObjectOutputStream(toServer.getOutputStream());
			out.writeObject(set);

			toServer.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
 	
	public void run() {
		// etsi opws exei ginei to dht i redistribute dn einai anagki
		// na kaleitai panta apo to node, mporei na trexei k peristasiaka
		for (;;) {
//			printChunks();
			redistributeKeys();
			try {
				Thread.sleep(5000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
	
	/**
	 * Is K between my predecessor and me ?
	 * @param k
	 * @return
	 */
	private boolean betweenPredAndMe(String k) {
		if ((pred.getHash().compareTo(k) < 0)
				&& (k.compareTo(node.getHash()) < 0)) {
			return true;
		}
		if ((pred.getHash().compareTo(k) < 0)
				&& (node.getHash().compareTo(pred.getHash()) < 0)) {
			return true;
		}
		if ((node.getHash().compareTo(pred.getHash()) < 0)
				&& (k.compareTo(node.getHash()) < 0)) {
			return true;
		} else {
			return false;
		}
	}

	private void printChunks() {
		for (Entry<String, FileChunk> e : dht.entrySet()) {
			Log.v("DHTEntry", e.getKey());
		}
		Log.v("DHT", "------------------------------");
	}
}
